package com.atguigu.gmall.realtime.app.dws;

import com.atguigu.gmall.realtime.app.func.KeywordUDTF;
import com.atguigu.gmall.realtime.bean.KeywordStats;
import com.atguigu.gmall.realtime.common.GmallConstant;
import com.atguigu.gmall.realtime.util.ClickHouseUtil;
import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author: xu
 * @desc: 搜索关键字
 * 程  序：mockLog -> Nginx -> Logger.sh -> Kafka(ZK) -> BaseLogApp -> kafka -> KeywordStatsApp -> ClickHouse
 */
public class KeywordStatsApp {
    public static void main(String[] args) throws Exception {
        // TODO 1.基本环境准备
        // 1.1  准备本地测试流环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 1.2 设置并行度
        env.setParallelism(1);
        // 1.3 设置Checkpoint
        // env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        // env.getCheckpointConfig().setCheckpointTimeout(60000);
        // env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000L));
        // env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // env.setStateBackend(new FsStateBackend("hdfs://node1:8020/gmall/checkpoint/KeywordStatsApp"))
        // System.setProperty("HADOOP_USER_NAME", "root");
        // 1.4 创建Table环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // TODO 2.注册自定义函数
        tableEnv.createTemporarySystemFunction("ik_analyze", KeywordUDTF.class);

        // TODO 3.创建动态表
        // 3.1 声明主题以及消费者组
        String pageViewSourceTopic = "dwd_page_log";
        String groupId = "keywordstats_app_group";
        // 3.2建表
        tableEnv.executeSql("CREATE TABLE page_view (" +
                " common MAP<STRING, STRING>," +
                " page MAP<STRING, STRING>," +
                " ts BIGINT," +
                " rowtime as TO_TIMESTAMP(FROM_UNIXTIME(ts/1000,'yyyy-MM-dd HH:mm:ss'))," +
                " WATERMARK FOR rowtime AS rowtime - INTERVAL '2' SECOND" +
                " ) WITH (" + MyKafkaUtil.getKafkaDDL(pageViewSourceTopic, groupId) + ")");

        // TODO 4.将动态表中表示搜索行为的记录过滤出来
        Table fullwordTable = tableEnv.sqlQuery("SELECT page['item'] fullword, rowtime" +
                " FROM page_view" +
                " WHERE page['page_id']='good_list' AND page['item'] IS NOT NULL"
        );
        // TODO 5.使用自定义的UDTF函数 对搜索关键词进行拆分
        Table keywordTable = tableEnv.sqlQuery("SELECT keyword, rowtime" +
                " FROM  " + fullwordTable + "," +
                " LATERAL TABLE(ik_analyze(fullword)) AS t(keyword)"
        );
        // TODO 6.分组、开窗、聚合计算
        Table reduceTable = tableEnv.sqlQuery("SELECT " +
                " DATE_FORMAT(TUMBLE_START(rowtime, INTERVAL '10' SECOND),'yyyy-MM-dd HH:mm:ss') as stt," +
                " DATE_FORMAT(TUMBLE_END(rowtime, INTERVAL '10' SECOND),'yyyy-MM-dd HH:mm:ss') as edt," +
                " keyword," +
                " COUNT(*) ct," +
                " '" + GmallConstant.KEYWORD_SEARCH + "' source," +
                " UNIX_TIMESTAMP() * 1000 as ts " +
                " FROM " + keywordTable +
                " GROUP BY TUMBLE(rowtime, INTERVAL '10' SECOND), keyword");

        // TODO 7.转换为流
        DataStream<KeywordStats> keywordStatsStream = tableEnv.toAppendStream(reduceTable, KeywordStats.class);
        keywordStatsStream.print(">>>>");

        // TODO 8.写入到ClickHouse
        keywordStatsStream.addSink(
                ClickHouseUtil.getJdbcSink("insert into keyword_stats (keyword,ct,source,stt,edt,ts) values(?,?,?,?,?,?)")
        );

        env.execute(KeywordStatsApp.class.getSimpleName());
    }
}
